package com.dc.dbus.session.client;

import java.util.Arrays;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import com.dc.dbus.session.Log;
import com.dc.dbus.session.RC;
import com.dc.dbus.session.client.bean.AsyncInfo;
import com.dc.dbus.session.client.bean.SyncInfo;
import com.dc.dbus.session.listener.ClientSideEventListener;
import com.dc.dbus.transport.commons.ClientSideReceiver;

public class ClientInvokeFeedback implements ClientSideReceiver {
	
	protected final ConcurrentHashMap<Long, Object> msgIdMapInvoke = new ConcurrentHashMap<Long, Object>();
	
	protected final ClientSideEventListener eventListener; 
	
	public ClientInvokeFeedback(ClientSideEventListener eventListener) {
		
		this.eventListener = eventListener;
		
		startCleanAsyncTimeOutMsgThread();
	}
	
	protected void startCleanAsyncTimeOutMsgThread() {
		
		Thread cleanAsyncTimeOutMsgThread = new Thread(new Runnable() {
			@Override
			public void run() {
				for (;;) {
					try {
						TimeUnit.MILLISECONDS.sleep(5000);
					} catch (InterruptedException e) {
						Log.sessionLogger.error("", e);
					};
					cleanAsyncTimeOutMsg();
				}
			}
		});
		cleanAsyncTimeOutMsgThread.setName("cleanAsyncTimeOutMsgThread");
		cleanAsyncTimeOutMsgThread.start();
	}
	
	@SuppressWarnings("rawtypes")
	public void cleanAsyncTimeOutMsg() {
		
		long timeNow = System.currentTimeMillis();
		
		for (Entry<Long, Object> entry : msgIdMapInvoke.entrySet()) {
			Object value = entry.getValue();
			if (value instanceof AsyncInfo) {
				AsyncInfo v = (AsyncInfo) value;
				if (timeNow > v.expireTime) {
					msgIdMapInvoke.remove(entry.getKey());
					v.future.failedByTimeOut();
				}
			}
			
		}
	}
	
	public SyncInfo syncInvoke(long msgId) {
		
		SyncInfo sync = new SyncInfo();
		msgIdMapInvoke.put(msgId, sync);
		
		return sync;
	}
	
	public SyncInfo syncInvokeTimeOut(long msgId) {
		return (SyncInfo) msgIdMapInvoke.remove(msgId);
	}
	
	@SuppressWarnings("rawtypes")
	public AsyncInfo asyncInvoke(long msgId, int timeout) {
		
		AsyncInfo async = new AsyncInfo(timeout + System.currentTimeMillis());
		msgIdMapInvoke.put(msgId, async);
		
		return async;
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	@Override
	public void handlerMsg(long replyForMsgId, int replyForCmd, int resultCode, Object param) {
		
		try {
			Object info = msgIdMapInvoke.remove(replyForMsgId);
			if (info == null) {
				throw new RuntimeException(
						"InvokeRecorder.handleMessage can't find invoke info for server back( id: " 
								+ replyForMsgId + ", cmd:" + replyForCmd + 
								", param:" + paramToString(param) + ")");
			} else {
				
				if (info instanceof SyncInfo) {
					
					SyncInfo sync = (SyncInfo) info;
					ReentrantLock lock = sync.lock;
					lock.lock();
					try {
						sync.resultCode = resultCode;
						sync.result = param;
						sync.condition.signal();
					} finally {
						lock.unlock();
					}
					
				} else {
					
					AsyncInfo async = (AsyncInfo) info;
					if (resultCode == RC.SUCCESS) {
						async.future.completed(param);
					} else {
						async.future.failedWithRemoteException(resultCode, (Throwable) param);
					}
				}
				
			}
		} catch (Exception e) {
			eventListener.exception("RabbitMqSender.handlerMsg Exception", e);
		}
	}
	
	protected String paramToString(Object param) {
		
		if (param instanceof Object[]) {
			return Arrays.toString((Object[])param);
		} else {
			return param.toString();
		}
	}
}
